package cn.zwx.remoting.netty;

import cn.zwx.config.ClientConfig;
import cn.zwx.config.FeignRpcProperties;
import cn.zwx.constant.FeignRpcCommonConstant;
import cn.zwx.enumerate.HandleEnum;
import cn.zwx.exception.RemotingConnectException;
import cn.zwx.exception.RemotingSendRequestException;
import cn.zwx.exception.RemotingTimeoutException;
import cn.zwx.remoting.RemotingCommand;
import com.alibaba.fastjson.JSON;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.tomcat.util.threads.TaskQueue;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * NettyClient端
 * @author 105286
 */
public class NettyRemotingClient extends AbstractNettyRemoting implements RemotingClient {

    private static final Logger logger = LoggerFactory.getLogger(NettyRemotingClient.class);

    /**
     *  存放唯一标示key（服务名称+"_"+端口号）和channel关系
     **/
    private ConcurrentHashMap<String, Channel> channelTables = new ConcurrentHashMap<>(8);
    /**
     *  bootstrap
     **/
    private Bootstrap bootstrap;
    /**
     *  工作线程
     **/
    private NioEventLoopGroup workGroup;
    /**
     *  工作线程下标
     **/
    private AtomicInteger workThreadIndex = new AtomicInteger(0);
    /**
     *  Boss线程下标
     **/
    private AtomicInteger clientHandleThreadIndex = new AtomicInteger(0);
    /**
     *  线程池
     **/
    private ThreadPoolExecutor poolExecutor;


    private FeignRpcProperties feignRpcProperties;

    public NettyRemotingClient(FeignRpcProperties feignRpcProperties) {
        this.feignRpcProperties = feignRpcProperties;
    }

    /**
     * 执行定时任务线程池
     */
    ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1,new BasicThreadFactory.Builder()
            .namingPattern(FeignRpcCommonConstant.CLIENT_HOUSE_KEEPING_SERVICE)
            .daemon(Boolean.TRUE)
            .priority( Thread.NORM_PRIORITY )
            .build());



    @Override
    public void start() {
        ClientConfig clientConfig = feignRpcProperties.getClient();

        TaskQueue taskQueue = new TaskQueue(clientConfig.getMaxTaskQueueSize());

        // 构建请求线程池
        poolExecutor = new ThreadPoolExecutor(clientConfig.getCoreThreadNum(), clientConfig.getMaxThreadNum(),  clientConfig.getMaxTheadFreeTime(), TimeUnit.SECONDS,taskQueue,
                r -> new Thread(r, FeignRpcCommonConstant.CLIENT_HANDLE_THREAD + clientHandleThreadIndex.getAndIncrement()));
        taskQueue.setParent(poolExecutor);

        workGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2, r -> {
            return new Thread(r, FeignRpcCommonConstant.NETTY_CLIENT_WORK_THREAD + workThreadIndex.getAndIncrement());
        });
        bootstrap = new Bootstrap();
        bootstrap.group(workGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_SNDBUF, 1024 * 1024 * 10)
                .option(ChannelOption.SO_RCVBUF, 1024 * 1024 * 10)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000 * 10)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new IdleStateHandler(0, 0, 30, TimeUnit.SECONDS));
                        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 0, 8, 0, 8));
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new RemotingClientHandler());
                        ch.pipeline().addLast(new RemotingCommandHandle());
                        ch.pipeline().addFirst(new StringEncoder());
                        ch.pipeline().addFirst(new LengthFieldPrepender(8));
                    }
                });
        // init scheduleAtFixedRate
        scheduledExecutorService.scheduleAtFixedRate(()->{
            try {
                //扫描结果响应表
                scanResponseTable();
            } catch (Exception e) {
                logger.error("scanResponseTable exception", e);
            }
        },1000 * 3L, 1000 * 10L,TimeUnit.MILLISECONDS);

    }


    @Override
    public RemotingCommand invokeSync(String addr, RemotingCommand command, long timeoutMillis) throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException, RemotingConnectException {
            Channel channel = getAndCheckChannel(addr);
            if (channel == null) {
                throw new RemotingConnectException(addr);
            }
            return super.invokeSyncImpl(channel, command, timeoutMillis);
    }


    private Channel getAndCheckChannel(String addr) {
        Channel channel = channelTables.get(addr);
        if (channel != null && !channel.isActive()) {
            channelTables.remove(addr);
            channel = null;
        }
        return channel;
    }

    /**
     *  跟服务端构建链接
     * @param ip ip
     * @param port 端口
     * @return void
     * @author zhangwenxue
     * @createTime 2022/6/20 14:40
     **/
    @Override
    public void connect(String ip, int port) {
        bootstrap.connect(ip, port).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                channelTables.put(ip + "_" + port, future.channel());
                if (logger.isInfoEnabled()){
                    logger.info("connect  to remoting server success, remoting address {}:{} ", ip, port);
                }
                return;
            }
            if (logger.isWarnEnabled()){
                logger.warn("connect to {}:{} fail,cause:{}", ip, port, future.cause().getMessage());
            }
        });
    }

    @Override
    public boolean channelActive(String key) {
        Channel channel = channelTables.get(key);
        if (channel == null) {
            return false;
        }
        return channel.isActive();
    }



    @Override
    public void shutdown() {
        if (workGroup != null) {
            workGroup.shutdownGracefully();
        }
        scheduledExecutorService.shutdown();
        if (!CollectionUtils.isEmpty(channelTables)) {
            for (Channel channel : channelTables.values()) {
                channel.close();
            }
        }
        channelTables.clear();
    }

    /**
     *  替除无效链接
     * @return void
     * @author zhangwenxue
     * @createTime 2022/6/20 15:54
     **/
    void removeUnActiveChannel() {
        List<String> keyList = new LinkedList<>();
        for (Map.Entry<String, Channel> entry : channelTables.entrySet()) {
            String key = entry.getKey();
            Channel value = entry.getValue();
            if (!value.isActive()) {
                keyList.add(key);
            }
        }
        for (String key : keyList) {
            channelTables.remove(key);
        }
    }

    private class RemotingClientHandler extends SimpleChannelInboundHandler<String> {

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state() == IdleState.ALL_IDLE) {
                    RemotingCommand ping = new RemotingCommand();
                    ping.setType(HandleEnum.PING_COMMAND.getCode());
                    ctx.pipeline().writeAndFlush(ping);
                }

            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            logger.error(cause.toString());
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
            poolExecutor.execute(() -> {
                RemotingCommand remotingCommand = JSON.parseObject(msg, RemotingCommand.class);
                processMessageReceived(ctx, remotingCommand);
            });


        }
    }
}
